package cn.mesmile.flink.source;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;

/**
 * @author zb
 * @date 2021/8/21 16:39
 * @Description
 */
public class FlinkSourceDemo03 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 自动运行环境模式
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        // 【元素集合】三种数据来源 fromCollection  fromElements   fromSequence
        DataStreamSource<String> stringDS1 = env.fromCollection(Arrays.asList("12", "34", "56,78"));
        stringDS1.print("stringDS1");

        DataStream<String> stringDS2 = env.fromElements("java,SpringBoot", "spring cloud,redis", "kafka,小滴课堂");
        stringDS2.print("stringDS2");

        DataStreamSource<Long> longDS3 = env.fromSequence(0,10);
        longDS3.print("longDS3");

        SingleOutputStreamOperator<String> stringSingleOutputStreamOperator = stringDS1.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] split = value.split(",");
                for (String val : split) {
                    out.collect(val);
                }
            }
        });

        stringSingleOutputStreamOperator.print("result_");
        env.execute("this is job name");
    }
}
